package com.gome.ss.flowcrawl.extractor;

import java.util.Set;

import javax.jms.JMSException;

import org.apache.log4j.Logger;
import org.bson.types.ObjectId;

import com.gome.ss.common.AppConfiguer;
import com.gome.ss.common.Utils;
import com.gome.ss.core.schedule.ThreadTaskBase;
import com.mongodb.BasicDBObject;
import com.mongodb.DBCollection;
import com.mongodb.DBCursor;
import com.mongodb.DBObject;

public class FlowCrawlExtractorFilterTask extends FlowCrawlExtractorTask implements ThreadTaskBase {
	private static Logger logger = Logger.getLogger(FlowCrawlExtractorFilterTask.class);
	
	private HTMLCount counter = null;
	private RemoveDuplicate removeDuplicate = new RemoveDuplicate();
	
	public FlowCrawlExtractorFilterTask(FlowCrawlExtractor extractor) {
		super(extractor);
		counter = new HTMLCount(FlowCrawlExtractorManager.getFilterDicFilePathName(extractor.getConfig().getName()));
	}
	
	@Override
	public void execute() {
		FlowCrawlExtractorConfig config = extractor.getConfig();
		
		DBCollection collection = AppConfiguer.flowCrawlConfigMongo.getDB(AppConfiguer.FLOWCRAWL_DB_NAME)
				 .getCollection(extractor.getConfig().getName());
		
		BasicDBObject query = new BasicDBObject();
		query.append("stepID", config.getStepID());
		DBCursor cursor = collection.find(query);
		
		int totalCount = cursor.count();
		int curpos = 0;
		
		logger.info(config.getName() + " filter task count " + cursor.count());
		while(cursor.hasNext() && !needStop) {
			DBObject obj = cursor.next();
			if (obj != null) {
				String pageSource = (String) obj.get("pageSource");
				if (pageSource != null) {
					ObjectId idObj = (ObjectId) obj.get("_id");
					String id = idObj.toString();
					try {
						removeDuplicate.putDoc(id, pageSource);
					} catch (Exception e) {
						logger.error(Utils.getTrace(e));
					}
				}
			}
			curpos++;
			if (curpos%100 == 0)
				logger.info(config.getName() + " filter removeDuplicate task " + curpos + " has end.");
		}
		
		logger.info(config.getName() + " filter getRemoveDupSet.");;
		Set<String> rd = removeDuplicate.getRemoveDupSet();
		logger.info(config.getName() + " filter getRemoveDupSet end.");;
		
		cursor = collection.find(query);
		curpos = 0;
		
		while(cursor.hasNext() && !needStop) {
			DBObject obj = cursor.next();
			if (obj != null) {
				
				curpos++;
				
				String pageSource = (String) obj.get("pageSource");
				if (pageSource != null) {
					ObjectId idObj = (ObjectId) obj.get("_id");
					String id = idObj.toString();
					if (rd.contains(id))
						continue;
					counter.count(pageSource);
				}
				
				if (curpos%100 == 0)
					logger.info(config.getName() + " filter counter task " + curpos + " has end.");
			}
		}
		
		counter.writeDict();
		extractor.loadFilterDict();
		
		logger.info(config.getName() + " filter task " + curpos + " end.");
		
		if (config.getEndCallbackLink() != null && !config.getEndCallbackLink().equals("")) {
			Utils.sendGet(config.getEndCallbackLink());
		}
		
		needStop = false;
		try {
			producer.close();
			session.close();
			connection.close();
		} catch (JMSException e) {
			logger.error(Utils.getTrace(e));
		}
		extractor.getStatus().stop();
	}
	
	@Override
	public void stop() {
		logger.info(extractor.getConfig().getName() + " filter task stop.");
		needStop = true;
	}
}